package com.itguigu.rabbitmq.four;

import com.itguigu.rabbitmq.utils.RabbitMqutil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

/**
 * @Date 2021/7/18 16:22
 * @Created by wangle
 * @Description 发布确认模式
 * 1.单个确认模式
 * 2.批量确认模式
 * 3.异步确认模式
 */
public class ConfirmMessage {

    public static final int message_count = 1000;

    public static void main(String[] args) throws Exception {

        //  ConfirmMessage.one(); //发布：1000条，耗时1960毫秒
        //ConfirmMessage.two(); //发布：1000条批量确认消息，耗时159毫秒
        ConfirmMessage.three();  //发布：1000条批量确认消息，耗时104毫秒

    }

    //单个确认
    public static void one() throws Exception {
        Channel channel = RabbitMqutil.getChannel();

        //队列名称
        String queueName = UUID.randomUUID().toString();
        //队列声明
        channel.queueDeclare(queueName, false, false, false, null);
        //开启发布确认
        channel.confirmSelect();
        //开启的时间
        Long begin = System.currentTimeMillis();
        //批量发消息
        for (int i = 0; i < message_count; i++) {

            String message = i + " ";
            channel.basicPublish("", queueName, null, message.getBytes(StandardCharsets.UTF_8));
            //单个消息就马上进行发布确认
            boolean flag = channel.waitForConfirms();
            if (flag) {

                System.out.println("消息发送成功！");
            }

        }

        //结束时间
        Long end = System.currentTimeMillis();
        System.out.println("发布：" + message_count + "条单个确认消息，耗时" + (end - begin) + "毫秒");

    }

    //批量确认
    public static void two() throws Exception {

        Channel channel = RabbitMqutil.getChannel();

        //队列名称
        String queueName = UUID.randomUUID().toString();
        //队列声明
        channel.queueDeclare(queueName, false, false, false, null);
        //开启发布确认
        channel.confirmSelect();
        //开启的时间
        Long begin = System.currentTimeMillis();
        //批量确认消息的大小
        int batchSize = 100;

        //未确认消息的个数

        //批量发消息,批量确认
        for (int i = 0; i < message_count; i++) {

            String message = i + " ";
            channel.basicPublish("", queueName, null, message.getBytes(StandardCharsets.UTF_8));
            //判断达到100条的时候确认一次
            boolean flag = false;
            if (i % 100 == 0) {

                flag = channel.waitForConfirms();
            }


            if (flag) {

                System.out.println("消息发送成功！");
            }

        }

        //结束时间
        Long end = System.currentTimeMillis();
        System.out.println("发布：" + message_count + "条批量确认消息，耗时" + (end - begin) + "毫秒");

    }

    //异步确认
    public static void three() throws Exception {


        Channel channel = RabbitMqutil.getChannel();

        //队列名称
        String queueName = UUID.randomUUID().toString();
        //队列声明
        channel.queueDeclare(queueName, false, false, false, null);
        //开启发布确认
        channel.confirmSelect();
        /*
         * 线程安全有序的一个哈希表，适用于高并发的情况下
         * 1.将序号与消息进行关联
         * 2.批量删除条目
         * 3.支持高并发（多线程）
         * */
        ConcurrentSkipListMap<Long, String> outstandingconfirms = new ConcurrentSkipListMap<>();

        //开启的时间
        Long begin = System.currentTimeMillis();

        //消息成功回调函数
        ConfirmCallback ackback = (deliverTag, multiple) -> {
            if (multiple) {
                //2.删除掉确认的消息
                ConcurrentNavigableMap confirmed = outstandingconfirms.headMap(deliverTag);
                //   outstandingconfirms.clear(confirmed);

                for (Object key : confirmed.keySet()
                ) {

                    outstandingconfirms.remove(key);
                }
                //  confirmed.clear();
            } else {

                outstandingconfirms.remove(deliverTag);
            }

            System.out.println("成功确认的消息" + deliverTag);

        };
        //消息失败回调函数
        ConfirmCallback nackback = (deliverTag, multiple) -> {

            //3.打印未确认的消息
            String message = outstandingconfirms.get(deliverTag);

            System.out.println("未确认的消息是：" + message + "未确认的消息" + deliverTag);

        };
        //准备消息的监听器，监听消息的成功和失败

        //1.成功的消息  2.失败的消息
        channel.addConfirmListener(ackback, nackback); //异步通知
        //批量发送消息
        for (int i = 0; i < message_count; i++) {
            String message = "消息" + i;
            channel.basicPublish("", queueName, null, message.getBytes(StandardCharsets.UTF_8));
            //1.记录要发布的消息总和
            outstandingconfirms.put(channel.getNextPublishSeqNo(), message);


        }


        //结束时间
        Long end = System.currentTimeMillis();
        System.out.println("发布：" + message_count + "条批量确认消息，耗时" + (end - begin) + "毫秒");

    }


}
